<?php

namespace App\Services\RabbitMQ;

use App\Services\BaseServices;
use Illuminate\Support\Facades\Log;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class LXServices extends BaseServices
{
    protected $connect;
    protected $channel;
    protected $config;

    public function __construct()
    {
        $this->config = config("queue.connections.rabbitmq");
        $this->connect = new AMQPStreamConnection(
            $this->config["host"],
            $this->config["port"],
            $this->config["login"],
            $this->config["password"],
            "/"
        );
        $this->channel = $this->connect->channel();
        $this->channel->exchange_declare($this->config["exchange_name"], "direct", false, true, false);//声明初始化交换机
        $this->channel->queue_declare($this->config["queue"], false, true, false, false);//声明初始化一条队列
        $this->channel->queue_bind($this->config["queue"], $this->config["exchange_name"]);//将队列与某个交换机进行绑定，并使用路由关键字
    }

    public function __destruct()
    {
        $this->channel->close();
        $this->connect->close();
    }

    /**
     * MQ生产者
     * @param  array  $data
     * @return bool
     */
    public function publishMq(array $data = [])
    {
        try {
            $json = json_encode($data);
            $msg = new AMQPMessage($json, ['content_type' => 'text/plain', 'delivery_mode' => 2]);//生成消息
            $this->channel->basic_publish($msg, $this->config["exchange_name"]);
            return true;
        } catch (\Exception $e) {
            Log::info("publishMQ: " + $e->getMessage());
            return false;
        }
    }

    /**
     * MQ消费者
     * @param  string  $queue
     * @param  bool  $bForceDelete  是否取后即删
     * @return AMQPMessage|null
     */
    public function consumerMq(string $queue, bool $bForceDelete = false)
    {
        try {
            $this->channel->queue_declare($queue, false, true, false, false);
            $message = $this->channel->basic_get($queue);
            if ($message && $bForceDelete) {
                $this->channel->basic_ack($message->delivery_info["delivery_tag"]);
            }
        } catch (\Exception $e) {
            Log::info($e->getMessage());
        }
        return $message;
    }
}
